//
// Created by dave on 18-7-28.
//

#pragma  once

#include <string>
#include <chrono>
#include <cassandra.h>
#include <glog/logging.h>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/io_context.hpp>
#include <libgo/coroutine.h>

#include "net/channel.h"
#include "net/jtt808_handler.h"

using boost::asio::ip::tcp;

class tcp_server {
public:
    tcp_server(boost::asio::io_context& serv_context, int port, CassSession* session)
            : serv_context_(serv_context), acceptor_(serv_context, tcp::endpoint(tcp::v4(), port)), strand_(serv_context),
            session_(session), dao_(session), saver_(std::bind(&location_dao::insert, &dao_, std::placeholders::_1)) { }

    void start() {
        accept();
        timer_ = std::make_shared<boost::asio::steady_timer>(serv_context_, std::chrono::microseconds(300));
        timer_->async_wait(std::bind(&tcp_server::housekeeper, this, std::placeholders::_1));
    }

    void accept() {
        acceptor_.async_accept(
                [this](boost::system::error_code ec, tcp::socket socket) {
                    if (!ec) {
                        socket.set_option(tcp::no_delay(true));
                        auto ptr = std::make_shared<channel>(std::move(socket), std::make_shared<jtt808_handler>(saver_, counter_), strand_);
                        go [ptr] {
                            ptr->start();
                        };
                    }
                    accept();
                });
    }

private:
    void housekeeper(boost::system::error_code ec) {
        saver_.flush();
        timer_->expires_from_now(std::chrono::microseconds(300));
        timer_->async_wait(std::bind(&tcp_server::housekeeper, this, std::placeholders::_1));
    }

private:
    boost::asio::io_context& serv_context_;
    boost::asio::io_context::strand strand_;
    tcp::acceptor acceptor_;
    CassSession* session_;
    location_dao dao_;
    pooled_data_saver<location> saver_;
    std::shared_ptr<boost::asio::steady_timer> timer_;
    std::atomic_int counter_{0};
};

